package com.cloud.easy.im.netty.channel;

import com.cloud.easy.im.netty.config.UDPServerChannelConfig;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.AbstractNioMessageChannel;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.ServerSocketChannelConfig;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.util.internal.PlatformDependent;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.StandardProtocolFamily;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.spi.SelectorProvider;
import java.util.LinkedHashMap;
import java.util.List;

/**
 * 仿照TCP协议的NioServerSocketChannel实现的专用于UDP的服务端Channel实现类。
 * {@link NioDatagramChannel} 类似netty原生提供的socket服务端类库
 *
 * @author wangjian
 * @version 0.1.0
 * @create 2020年09月02日6:38 下午
 */
public final class UDPServerChannel extends AbstractNioMessageChannel implements ServerSocketChannel {

    /**
     * channel配置元信息，当为true时允许用户断开链接
     **/
    private final ChannelMetadata METADATA = new ChannelMetadata(true);

    private final ServerSocketChannelConfig config;

    protected final LinkedHashMap<InetSocketAddress, UDPClientChannel> clientChannels = new LinkedHashMap<InetSocketAddress, UDPClientChannel>();

    public UDPServerChannel() throws IOException {
        //SelectorProvider.provider()    .openDatagramChannel		   		   StandardProtocolFamily.INET
        //返回系统范围内的默认选择器提供程序 ｜ 打开并创建一个操作系统支持的UDP channel ｜ 互联网协议版本4（IPv4）
        this(SelectorProvider.provider().openDatagramChannel(StandardProtocolFamily.INET));
    }

    protected UDPServerChannel(DatagramChannel datagramChannel) throws IOException {
        //构建底层NIO channel通道实例 @{see io.netty.channel.nio.AbstractNioChannel.AbstractNioChannel}
        super(null, datagramChannel, SelectionKey.OP_READ);
        this.config = new UDPServerChannelConfig(this, datagramChannel);
    }

    @Override
    public InetSocketAddress localAddress() {
        return (InetSocketAddress) super.localAddress();
    }

    @Override
    protected SocketAddress localAddress0() {
        return this.javaChannel().socket().getLocalSocketAddress();
    }

    @Override
    public InetSocketAddress remoteAddress() {
        return null;
    }

    @Override
    protected SocketAddress remoteAddress0() {
        return null;
    }

    @Override
    public ChannelMetadata metadata() {
        return METADATA;
    }

    @Override
    public ServerSocketChannelConfig config() {
        return config;
    }

    /**
     * 判断监听是否已启动。
     *
     * @return true表示已已启动监听，否则未启动
     * @see DatagramChannel#isOpen()
     */
    @Override
    public boolean isActive() {
        return this.javaChannel().isOpen() && this.javaChannel().socket().isBound();
    }

    @Override
    protected DatagramChannel javaChannel() {
        return (DatagramChannel) super.javaChannel();
    }

    @Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        javaChannel().socket().bind(localAddress);
    }

    @Override
    protected void doClose() throws Exception {
        for (UDPClientChannel channel : clientChannels.values()) {
            channel.close();
        }

        javaChannel().close();
    }

    /**
     * 将一个客户端的Channel实例从服务端管理的列表中移除。
     *
     * @param channel 服务端channel
     */
    public void removeChannel(final Channel channel) {
        eventLoop().submit(new Runnable() {
            @Override
            public void run() {
                InetSocketAddress remote = (InetSocketAddress) channel.remoteAddress();
                if (clientChannels.get(remote) == channel) {
                    clientChannels.remove(remote);
                }
            }
        });
    }

    @Override
    protected int doReadMessages(List<Object> list) throws Exception {
        //获取java原生nio提供UDP的channel
        DatagramChannel javaChannel = javaChannel();
        //获取一个自适应的缓冲区分配器
        RecvByteBufAllocator.Handle allocatorHandle = unsafe().recvBufAllocHandle();
        //分配一个缓冲
        ByteBuf buf = allocatorHandle.allocate(config.getAllocator());
        //将通道中的数据读取到缓冲中
        allocatorHandle.attemptedBytesRead(buf.writableBytes());

        boolean freeBuffer = true;
        try {
            // 将ByteBuf中可写的ByteBuffer取出
            ByteBuffer nioBuffer = buf.internalNioBuffer(buf.writerIndex(), buf.writableBytes());
            //获取链接信息之前ByteBuffer写入位置
            // ======｜=============|
            //       ^			    ^
            //     position     capacity
            int nioPos = nioBuffer.position();

            // 获取客户端链接ip和port
            // ======｜====IP/port===|===========|
            //       ^		      	 ^	         ^
            //     nioPos	     position     capacity
            InetSocketAddress inetSocketAddress = (InetSocketAddress) javaChannel.receive(nioBuffer);
            if (inetSocketAddress == null) {
                return 0;
            }

            //设置上次读取操作已读取的字节。可用于增加以读取的字节数
            allocatorHandle.lastBytesRead(nioBuffer.position() - nioPos);

            buf.writerIndex(buf.writerIndex() + allocatorHandle.lastBytesRead());

            // 分配新channel或使用现有channel并将消息推送到该channel
            UDPClientChannel udpchannel = clientChannels.get(inetSocketAddress);
            //如果address对应channel为空，则创建新channel
            if ((udpchannel == null) || !udpchannel.isOpen()) {
                udpchannel = new UDPClientChannel(this, inetSocketAddress);
                clientChannels.put(inetSocketAddress, udpchannel);
                list.add(udpchannel);

                udpchannel.addBuffer(buf);
                freeBuffer = false;

                return 1;
            } else {
                udpchannel.addBuffer(buf);
                freeBuffer = false;

                if (udpchannel.isRegistered()) {
                    udpchannel.read();
                }

                return 0;
            }
        } catch (Throwable t) {
            PlatformDependent.throwException(t);
            return -1;
        } finally {
            if (freeBuffer) {
                // 如果属于无用buffer，需要即使回收
                buf.release();
            }
        }
    }

    @Override
    protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer buffer) throws Exception {
        DatagramPacket dpacket = (DatagramPacket) msg;
        InetSocketAddress recipient = dpacket.recipient();
        ByteBuf byteBuf = dpacket.content();
        int readableBytes = byteBuf.readableBytes();
        if (readableBytes == 0) {
            return true;
        }

        ByteBuffer internalNioBuffer = byteBuf.internalNioBuffer(
                byteBuf.readerIndex(), readableBytes);

        return javaChannel().send(internalNioBuffer, recipient) > 0;
    }

    @Override
    protected boolean doConnect(SocketAddress addr1, SocketAddress addr2) throws Exception {
        throw new UnsupportedOperationException();
    }

    @Override
    protected void doFinishConnect() throws Exception {
        throw new UnsupportedOperationException();
    }

    @Override
    protected void doDisconnect() throws Exception {
        throw new UnsupportedOperationException();
    }
}
